/*
 * SocketChannelFactory.java
 *
 * Created on Jun 9, 2007, 10:33:46 AM
 *
 * To change this template, choose Tools | Template Manager
 * and open the template in the editor.
 */

package com.infomancers.workflow.tests;

import com.infomancers.workflow.Token;
import com.infomancers.workflow.channels.ChannelFactory;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 *
 * @author aviadbd
 */
public class SocketChannelFactory extends ChannelFactory {
    
    private int port;
    
    public SocketChannelFactory(int port) {
        super(Executors.newFixedThreadPool(5));
        this.port = port;
    }
    
    protected void internalSend(Token token) throws IOException {
        Socket socket = new Socket("localhost", port);
        
        ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
        oos.writeObject(token);
    }
    
    protected Token internalRetrieve() throws IOException {
        try {
            return retrievedTokens.take();
        } catch (InterruptedException ex) {
            Logger.getLogger("global").log(Level.SEVERE, null, ex);
            return null;
        }
    }
    
    private BlockingQueue<Token> retrievedTokens = new LinkedBlockingQueue<Token>();
    private ServerSocket server;
    private boolean closedProperly = false;
    
    protected void initInputChannel() throws IOException { }

    protected void initOutputChannel() throws IOException {
        server = new ServerSocket(port);
        Runnable retriever = new Runnable() {
            public void run() {
                while (!closedProperly) {
                    try {
                        java.net.Socket socket = server.accept();
                        java.io.ObjectInputStream ois = new java.io.ObjectInputStream(socket.getInputStream());
                        Token token = (Token) ois.readObject();
                        retrievedTokens.put(token);
                    } catch (IOException ex) {
                        if (!closedProperly) {
                            Logger.getLogger("global").log(Level.SEVERE, "Transmit layer error", ex);
                        }
                    } catch (ClassNotFoundException ex) {
                        Logger.getLogger("global").log(Level.SEVERE, "Couldn't find data class for token", ex);
                    } catch (InterruptedException ex) {
                        Logger.getLogger("global").log(Level.SEVERE, "Waiting on queue interrupted", ex);
                    }
                }
            }
            
        };
        
        getExecutor().execute(retriever);
    }
    
    public void close() throws IOException {
        if (server != null) {
            closedProperly = true;
            server.close();
        }
    }
    
}
